<?php
/**
 * Redis 5.0 引入了 Stream 数据结构，提供更强大的消息队列功能，支持持久化和消息确认。
 * 更高效的任务队列管理
 * 基本流程
 * 生产者：使用 XADD 将消息添加到 Stream。 xAdd('mystream', '*', $d);
 * 消费者：使用 XREAD 或 XREADGROUP 读取消息。 xRead(['mystream' => '$'], 1, 0)
 *
 * 添加数据 php mq/stream.php mq
 * 消费数据 php mq/stream.php
 */

class MqList
{
    /**
     * 添加数据
     */
    static function mq($data = ['task_id' => 1, 'message' => 'Stream Queue Example'])
    {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        $data['task_id'] = date('YmdHis');
        //$d = json_encode($data);
        $d = $data;
        $redis->xAdd('mystream', '*', $d);    // '*' 表示自动生成ID
        echo "Message added to stream.\n";
    }

    /**
     * 消费数据
     */
    static function server()
    {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        while (true) {
            // '$' 表示只读取新消息; '0-0' 从头开始读,就是下标为第一条的数据
            //block = 0 永久阻塞，直到有新消息 等待的毫秒数
            $messages = $redis->xRead(['mystream' => '$'], 1, 0);
            if ($messages) {
                //$stream  就是key=mystream
                foreach ($messages as $stream => $entries) {
                    foreach ($entries as $id => $fields) {
                        echo "Received message ID {$id}: " . json_encode($fields) . "\n";
                    }
                }
            }
        }
    }
}

if ('mq' == ($argv[1] ?? '') || 'php-fpm' == PHP_SAPI) {
    MqList::mq();
    return;
}

MqList::server();







